/**
 * Copyright (C), 2020-2020, 软件公司
 * FileName: RabbitConsumer.java
 * Author:   cakin
 * Date:     2020/5/2
 * Description: 消费端
 */
package com.rabbitmq.basicqos;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.util.ConnectionUtils;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static com.rabbitmq.common.Constant.NUM5;

/**
 * Qos测试的消费端
 */
@Slf4j
public class QosConsumer {
    /**
     * 消息队列
     */
    private static final String QUEUE_NAME = "queue.basicqos";

    /**
     * 控制ack消息
     */
    private static int num = 0;

    public static void main( String[] args ) {
        try {
            Connection connection = ConnectionUtils.getConnection();
            Channel channel = connection.createChannel();
            // 客户端最多未确认消息数为5，一旦超过5，客户端将无法接收到消息
            channel.basicQos(NUM5);
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery( String consumerTag,
                                            Envelope envelope,
                                            AMQP.BasicProperties properties,
                                            byte[] body )
                        throws IOException {
                    log.info(" recv message: " + new String(body));
                    // 控制是接收到的第几个消息
                    num++;
                    // 模拟业务处理耗时1秒
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                        log.info("catch InterruptedException");
                    }

                    // 只让是5的倍数的消息发ack消息
                    if (num % NUM5 == 0) {
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    } else {
                        log.trace("模拟不发ack情况");
                        // channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            channel.basicConsume(QUEUE_NAME, consumer);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
    }
}
